media_pp\elements\source\compositor/
video_compositor.rs

1use std::{
2    collections::{HashMap, HashSet},
3    sync::{
4        Arc, Mutex, Weak,
5        atomic::{AtomicU64, Ordering},
6    },
7    thread,
8    time::{Duration, Instant},
9};
10
11use crate::pp_log::{PpLog, pp_info};
12use arc_swap::ArcSwapOption;
13use ffmpeg_next as ffmpeg;
14use thiserror::Error as ThisError;
15
16use super::video_layer::{
17    self, LayerGeometry, MAX_DIMENSION, VideoFit, VideoInputId, VideoLayer, VideoLayerError,
18    VideoRect,
19};
20use crate::{
21    buffer::MediaBuffer,
22    bus::{Bus, BusEvent},
23    color::Color,
24    control::{ControlMsg, ControlReceiver, drain_control},
25    element::{Element, ElementType, Sink, Source, SourceElement, element_pp_log},
26    error::Result,
27    pad::SrcPad,
28    pool::{UnboundObjectPool, UnboundObjectPoolRef},
29    schedule::PeriodicSchedule,
30};
31
32const OUTPUT_POOL_SIZE: usize = 4;
33const CONTROL_POLL_INTERVAL: Duration = Duration::from_millis(5);
34
35/// The compositor's fixed output definition. Every emitted frame is an
36/// opaque [`ffmpeg::format::Pixel::BGRA`] frame at `width` x `height`.
37#[derive(Debug, Clone, Copy)]
38pub struct VideoCompositorOptions {
39    pub width: u32,
40    pub height: u32,
41    pub frame_rate: ffmpeg::Rational,
42    pub background: Color,
43}
44
45impl Default for VideoCompositorOptions {
46    fn default() -> Self {
47        Self {
48            width: 1920,
49            height: 1080,
50            frame_rate: ffmpeg::Rational::new(30, 1),
51            background: Color::BLACK,
52        }
53    }
54}
55
56/// Errors specific to [`VideoCompositor`].
57#[derive(Debug, ThisError)]
58pub enum VideoCompositorError {
59    #[error("ffmpeg error: {0}")]
60    Ffmpeg(#[from] ffmpeg::Error),
61
62    #[error(
63        "invalid output dimensions {width}x{height}; each dimension must be 1..={MAX_DIMENSION}"
64    )]
65    InvalidOutputDimensions { width: u32, height: u32 },
66
67    #[error("invalid frame rate {0}; numerator and denominator must both be positive")]
68    InvalidFrameRate(ffmpeg::Rational),
69
70    #[error(
71        "invalid layer dimensions {width}x{height}; each dimension must be 1..={MAX_DIMENSION}"
72    )]
73    InvalidLayerDimensions { width: u32, height: u32 },
74
75    #[error("layer opacity must be finite and between 0.0 and 1.0, got {0}")]
76    InvalidOpacity(f32),
77
78    #[error("input frame has invalid dimensions {width}x{height}")]
79    InvalidInputDimensions { width: u32, height: u32 },
80
81    #[error("scaled layer would exceed {MAX_DIMENSION}px: {width}x{height}")]
82    ScaledLayerTooLarge { width: u32, height: u32 },
83
84    #[error("the compositor input has been removed")]
85    SourceRemoved,
86
87    #[error(
88        "VideoCompositorInputSink only accepts decoded Video frames, got a {0}; link it after a decoder or video source"
89    )]
90    UnsupportedBuffer(&'static str),
91
92    #[error("VideoCompositor doesn't support seeking a live composition")]
93    SeekUnsupported,
94}
95
96struct VideoInput {
97    id: VideoInputId,
98    /// The hot producer/consumer path is an atomic latest-value slot:
99    /// input pipelines replace the pointer without taking the layer lock,
100    /// and the compositor acquires a stable Arc snapshot independently.
101    latest_frame: ArcSwapOption<UnboundObjectPoolRef<ffmpeg::frame::Video>>,
102    /// Layer changes are infrequent and update several related fields as
103    /// one coherent value, so a small dedicated lock remains appropriate.
104    layer: Mutex<VideoLayer>,
105}
106
107struct CompositorShared {
108    inputs: Mutex<HashMap<Arc<str>, Arc<VideoInput>>>,
109    next_input_id: AtomicU64,
110}
111
112/// A cheaply cloneable handle for adding and removing compositor inputs.
113/// It mirrors [`crate::elements::MixerHandle`], but each registration also
114/// returns a [`VideoLayerHandle`] for changing that input's placement.
115#[derive(Clone)]
116pub struct VideoCompositorHandle {
117    shared: Weak<CompositorShared>,
118}
119
120/// The two endpoints created for one compositor input registration.
121/// Move `sink` into the upstream pipeline and retain `layer` in application
122/// code for runtime placement changes.
123pub struct VideoCompositorInput {
124    pub sink: Box<dyn Sink>,
125    pub layer: VideoLayerHandle,
126}
127
128impl VideoCompositorHandle {
129    /// Registers an input and returns its terminal Sink plus independent
130    /// runtime layer control. Reusing `name` replaces the old registration;
131    /// old sinks and layer handles become harmlessly stale.
132    pub fn add_source(
133        &self,
134        name: impl Into<String>,
135        layer: VideoLayer,
136    ) -> std::result::Result<Option<VideoCompositorInput>, VideoCompositorError> {
137        validate_layer(layer)?;
138        let Some(shared) = self.shared.upgrade() else {
139            return Ok(None);
140        };
141        let name: Arc<str> = name.into().into();
142        let id = VideoInputId(shared.next_input_id.fetch_add(1, Ordering::Relaxed));
143        let input = Arc::new(VideoInput {
144            id,
145            latest_frame: ArcSwapOption::empty(),
146            layer: Mutex::new(layer),
147        });
148        shared
149            .inputs
150            .lock()
151            .unwrap()
152            .insert(name.clone(), input.clone());
153
154        Ok(Some(VideoCompositorInput {
155            sink: Box::new(VideoCompositorInputSink {
156                name: name.clone(),
157                pp_log: element_pp_log(ElementType::VideoCompositor, &name, None),
158                shared: self.shared.clone(),
159                input: Arc::downgrade(&input),
160            }),
161            layer: VideoLayerHandle {
162                id,
163                name,
164                input: Arc::downgrade(&input),
165            },
166        }))
167    }
168
169    /// Removes `name` immediately. Existing input sinks and layer handles
170    /// become disconnected and cannot affect a later same-name source.
171    pub fn remove_source(&self, name: &str) {
172        if let Some(shared) = self.shared.upgrade() {
173            shared.inputs.lock().unwrap().remove(name);
174        }
175    }
176
177    pub fn source_count(&self) -> usize {
178        self.shared
179            .upgrade()
180            .map(|shared| shared.inputs.lock().unwrap().len())
181            .unwrap_or(0)
182    }
183}
184
185/// Thread-safe runtime placement control for one compositor input.
186/// Retaining it does not keep the input or compositor alive.
187#[derive(Clone)]
188pub struct VideoLayerHandle {
189    id: VideoInputId,
190    name: Arc<str>,
191    input: Weak<VideoInput>,
192}
193
194impl VideoLayerHandle {
195    pub fn id(&self) -> VideoInputId {
196        self.id
197    }
198
199    pub fn name(&self) -> Arc<str> {
200        self.name.clone()
201    }
202
203    pub fn layer(&self) -> Option<VideoLayer> {
204        self.input
205            .upgrade()
206            .map(|input| *input.layer.lock().unwrap())
207    }
208
209    pub fn set_layer(&self, layer: VideoLayer) -> std::result::Result<(), VideoCompositorError> {
210        validate_layer(layer)?;
211        self.update(|current| *current = layer)
212    }
213
214    pub fn set_rect(&self, rect: VideoRect) -> std::result::Result<(), VideoCompositorError> {
215        validate_rect(rect)?;
216        self.update(|layer| layer.rect = rect)
217    }
218
219    pub fn set_opacity(&self, opacity: f32) -> std::result::Result<(), VideoCompositorError> {
220        validate_opacity(opacity)?;
221        self.update(|layer| layer.opacity = opacity)
222    }
223
224    pub fn set_z_index(&self, z_index: i32) -> std::result::Result<(), VideoCompositorError> {
225        self.update(|layer| layer.z_index = z_index)
226    }
227
228    pub fn set_visible(&self, visible: bool) -> std::result::Result<(), VideoCompositorError> {
229        self.update(|layer| layer.visible = visible)
230    }
231
232    pub fn set_fit(&self, fit: VideoFit) -> std::result::Result<(), VideoCompositorError> {
233        self.update(|layer| layer.fit = fit)
234    }
235
236    fn update(
237        &self,
238        update: impl FnOnce(&mut VideoLayer),
239    ) -> std::result::Result<(), VideoCompositorError> {
240        let input = self
241            .input
242            .upgrade()
243            .ok_or(VideoCompositorError::SourceRemoved)?;
244        update(&mut input.layer.lock().unwrap());
245        Ok(())
246    }
247}
248
249/// One terminal video input returned by
250/// [`VideoCompositorHandle::add_source`]. It stores only the latest frame,
251/// so a fast producer cannot build an unbounded queue behind a slower
252/// compositor output rate.
253pub struct VideoCompositorInputSink {
254    pp_log: PpLog,
255    name: Arc<str>,
256    shared: Weak<CompositorShared>,
257    input: Weak<VideoInput>,
258}
259
260impl VideoCompositorInputSink {
261    fn detach(&self) {
262        let (Some(shared), Some(input)) = (self.shared.upgrade(), self.input.upgrade()) else {
263            return;
264        };
265        let mut inputs = shared.inputs.lock().unwrap();
266        let is_current = inputs
267            .get(&self.name)
268            .is_some_and(|current| Arc::ptr_eq(current, &input));
269        if is_current {
270            inputs.remove(&self.name);
271        }
272    }
273}
274
275impl Element for VideoCompositorInputSink {
276    fn name(&self) -> Arc<str> {
277        self.name.clone()
278    }
279
280    fn element_type(&self) -> ElementType {
281        ElementType::VideoCompositor
282    }
283
284    fn pp_log(&self) -> &PpLog {
285        &self.pp_log
286    }
287
288    fn pp_log_mut(&mut self) -> &mut PpLog {
289        &mut self.pp_log
290    }
291}
292
293impl Sink for VideoCompositorInputSink {
294    fn consume(&mut self, buf: MediaBuffer) -> Result<()> {
295        let Some(input) = self.input.upgrade() else {
296            return Ok(());
297        };
298        match buf {
299            MediaBuffer::Video(frame) => {
300                input.latest_frame.store(Some(frame));
301                Ok(())
302            }
303            MediaBuffer::Eos => {
304                self.detach();
305                Ok(())
306            }
307            MediaBuffer::Packet(_) => Err(VideoCompositorError::UnsupportedBuffer("Packet").into()),
308            MediaBuffer::Audio(_) => Err(VideoCompositorError::UnsupportedBuffer("Audio").into()),
309        }
310    }
311
312    fn control(&mut self, msg: ControlMsg) -> Result<()> {
313        match msg {
314            ControlMsg::Stop => self.detach(),
315            ControlMsg::Seek(_) => {
316                if let Some(input) = self.input.upgrade() {
317                    input.latest_frame.store(None);
318                }
319            }
320            ControlMsg::Pause | ControlMsg::Resume => {}
321        }
322        Ok(())
323    }
324}
325
326#[derive(Clone)]
327struct InputSnapshot {
328    id: VideoInputId,
329    layer: VideoLayer,
330    frame: Option<Arc<UnboundObjectPoolRef<ffmpeg::frame::Video>>>,
331}
332
333#[derive(Debug, Clone, Copy, PartialEq, Eq)]
334struct ScaleDefinition {
335    source_format: ffmpeg::format::Pixel,
336    source_width: u32,
337    source_height: u32,
338    target_width: u32,
339    target_height: u32,
340}
341
342struct InputScaler {
343    definition: Option<ScaleDefinition>,
344    context: Option<ffmpeg::software::scaling::Context>,
345    output: ffmpeg::frame::Video,
346}
347
348impl InputScaler {
349    fn new() -> Self {
350        Self {
351            definition: None,
352            context: None,
353            output: ffmpeg::frame::Video::empty(),
354        }
355    }
356
357    fn scale(
358        &mut self,
359        source: &ffmpeg::frame::Video,
360        width: u32,
361        height: u32,
362    ) -> std::result::Result<&ffmpeg::frame::Video, ffmpeg::Error> {
363        let definition = ScaleDefinition {
364            source_format: source.format(),
365            source_width: source.width(),
366            source_height: source.height(),
367            target_width: width,
368            target_height: height,
369        };
370        if self.definition != Some(definition) {
371            self.context = Some(ffmpeg::software::scaling::Context::get(
372                definition.source_format,
373                definition.source_width,
374                definition.source_height,
375                ffmpeg::format::Pixel::BGRA,
376                width,
377                height,
378                ffmpeg::software::scaling::Flags::BILINEAR,
379            )?);
380            self.output = ffmpeg::frame::Video::new(ffmpeg::format::Pixel::BGRA, width, height);
381            self.definition = Some(definition);
382        }
383        self.context
384            .as_mut()
385            .expect("created with a changed definition or retained from a matching one")
386            .run(source, &mut self.output)?;
387        Ok(&self.output)
388    }
389}
390
391/// Composites the latest frames from any number of independent input
392/// pipelines into one fixed-rate opaque BGRA video stream.
393///
394/// Like [`crate::elements::AudioMixer`], this is a [`SourceElement`], not
395/// a conventional one-input filter: upstream pipelines terminate at the
396/// sinks returned by [`VideoCompositorHandle::add_source`], while this
397/// element's own pipeline drives output on its independent clock. Input
398/// frame PTS values therefore do not become output PTS; output advances by
399/// one tick in [`VideoCompositor::time_base`] for every composed frame.
400pub struct VideoCompositor {
401    pp_log: PpLog,
402    name: Arc<str>,
403    shared: Arc<CompositorShared>,
404    options: VideoCompositorOptions,
405    frame_interval: Duration,
406    frame_index: i64,
407    scalers: HashMap<VideoInputId, InputScaler>,
408    output_pool: UnboundObjectPool<ffmpeg::frame::Video>,
409    pad: SrcPad,
410}
411
412// SAFETY: `SwsContext` has no thread affinity and every scaling context is
413// exclusively accessed through `&mut self` on the compositor's one source
414// thread. ffmpeg-next simply omits Send for this wrapper, as with Scaler.
415unsafe impl Send for VideoCompositor {}
416
417impl VideoCompositor {
418    pub fn new(
419        name: impl Into<String>,
420        options: VideoCompositorOptions,
421    ) -> std::result::Result<(Self, VideoCompositorHandle), VideoCompositorError> {
422        validate_output_options(options)?;
423        let name: Arc<str> = name.into().into();
424        let pp_log = element_pp_log(ElementType::VideoCompositor, &name, None);
425        let shared = Arc::new(CompositorShared {
426            inputs: Mutex::new(HashMap::new()),
427            next_input_id: AtomicU64::new(1),
428        });
429        let frame_interval = Duration::from_secs_f64(
430            options.frame_rate.denominator() as f64 / options.frame_rate.numerator() as f64,
431        );
432        let (width, height) = (options.width, options.height);
433        let output_pool = UnboundObjectPool::new(
434            OUTPUT_POOL_SIZE,
435            move || ffmpeg::frame::Video::new(ffmpeg::format::Pixel::BGRA, width, height),
436            |_| {},
437        );
438        pp_info!(
439            pp_log: &pp_log,
440            "created: {}x{}, frame_rate={}, format=BGRA",
441            width,
442            height,
443            options.frame_rate
444        );
445        Ok((
446            Self {
447                name: name.clone(),
448                pp_log,
449                shared: shared.clone(),
450                options,
451                frame_interval,
452                frame_index: 0,
453                scalers: HashMap::new(),
454                output_pool,
455                pad: SrcPad::new(format!("{name}_src")),
456            },
457            VideoCompositorHandle {
458                shared: Arc::downgrade(&shared),
459            },
460        ))
461    }
462
463    pub fn format(&self) -> ffmpeg::format::Pixel {
464        ffmpeg::format::Pixel::BGRA
465    }
466
467    pub fn width(&self) -> u32 {
468        self.options.width
469    }
470
471    pub fn height(&self) -> u32 {
472        self.options.height
473    }
474
475    pub fn frame_rate(&self) -> ffmpeg::Rational {
476        self.options.frame_rate
477    }
478
479    pub fn time_base(&self) -> ffmpeg::Rational {
480        ffmpeg::Rational::new(
481            self.options.frame_rate.denominator(),
482            self.options.frame_rate.numerator(),
483        )
484    }
485
486    fn snapshots(&self) -> Vec<InputSnapshot> {
487        let inputs: Vec<_> = self
488            .shared
489            .inputs
490            .lock()
491            .unwrap()
492            .values()
493            .cloned()
494            .collect();
495        inputs
496            .into_iter()
497            .map(|input| InputSnapshot {
498                id: input.id,
499                layer: *input.layer.lock().unwrap(),
500                frame: input.latest_frame.load_full(),
501            })
502            .collect()
503    }
504
505    fn compose_frame(
506        &mut self,
507    ) -> std::result::Result<UnboundObjectPoolRef<ffmpeg::frame::Video>, VideoCompositorError> {
508        let mut snapshots = self.snapshots();
509        let active: HashSet<_> = snapshots.iter().map(|snapshot| snapshot.id).collect();
510        self.scalers.retain(|id, _| active.contains(id));
511        snapshots.sort_by(|left, right| {
512            left.layer
513                .z_index
514                .cmp(&right.layer.z_index)
515                .then_with(|| left.id.cmp(&right.id))
516        });
517
518        let mut output = self.output_pool.get();
519        fill_background(&mut output, self.options.background);
520        for snapshot in snapshots {
521            if !snapshot.layer.visible || snapshot.layer.opacity == 0.0 {
522                continue;
523            }
524            let Some(frame) = snapshot.frame else {
525                continue;
526            };
527            let geometry = layer_geometry(
528                frame.width(),
529                frame.height(),
530                snapshot.layer.rect,
531                snapshot.layer.fit,
532            )?;
533            let scaled = self
534                .scalers
535                .entry(snapshot.id)
536                .or_insert_with(InputScaler::new)
537                .scale(&frame, geometry.image_width, geometry.image_height)
538                .map_err(VideoCompositorError::from)?;
539            blend_bgra(&mut output, scaled, geometry, snapshot.layer.opacity);
540        }
541        output.set_pts(Some(self.frame_index));
542        self.frame_index += 1;
543        Ok(output)
544    }
545
546    fn push_frame(&mut self, bus: &Bus) -> std::result::Result<(), VideoCompositorError> {
547        let output = self.compose_frame()?;
548        if let Err(error) = self.pad.push(MediaBuffer::Video(Arc::new(output))) {
549            bus.post(
550                &self.pp_log,
551                BusEvent::Error {
552                    element_type: ElementType::VideoCompositor,
553                    name: self.name.clone(),
554                    error,
555                },
556            );
557        }
558        Ok(())
559    }
560}
561
562impl Element for VideoCompositor {
563    fn name(&self) -> Arc<str> {
564        self.name.clone()
565    }
566
567    fn element_type(&self) -> ElementType {
568        ElementType::VideoCompositor
569    }
570
571    fn pp_log(&self) -> &PpLog {
572        &self.pp_log
573    }
574
575    fn pp_log_mut(&mut self) -> &mut PpLog {
576        &mut self.pp_log
577    }
578}
579
580impl Source for VideoCompositor {
581    fn src_pads(&mut self) -> &mut [SrcPad] {
582        std::slice::from_mut(&mut self.pad)
583    }
584}
585
586impl SourceElement for VideoCompositor {
587    fn run(&mut self, control: &ControlReceiver, bus: &Bus) -> Result<()> {
588        pp_info!(self, "started");
589        let mut schedule = PeriodicSchedule::new(self.frame_interval, Instant::now());
590        loop {
591            let outcome = drain_control(control, self, bus)?;
592            if outcome.stopped {
593                pp_info!(self, "stopped");
594                return Ok(());
595            }
596            if outcome.paused_for > Duration::ZERO {
597                schedule.resume_after_pause(outcome.paused_for, Instant::now());
598            }
599
600            let now = Instant::now();
601            if !schedule.is_due(now) {
602                thread::sleep(schedule.remaining(now).min(CONTROL_POLL_INTERVAL));
603                continue;
604            }
605
606            self.push_frame(bus)?;
607            schedule.advance_after_tick(Instant::now());
608        }
609    }
610
611    fn seek(&mut self, _target: Duration) -> Result<Duration> {
612        Err(VideoCompositorError::SeekUnsupported.into())
613    }
614}
615
616fn validate_output_options(
617    options: VideoCompositorOptions,
618) -> std::result::Result<(), VideoCompositorError> {
619    if options.width == 0
620        || options.height == 0
621        || options.width > MAX_DIMENSION
622        || options.height > MAX_DIMENSION
623    {
624        return Err(VideoCompositorError::InvalidOutputDimensions {
625            width: options.width,
626            height: options.height,
627        });
628    }
629    if options.frame_rate.numerator() <= 0 || options.frame_rate.denominator() <= 0 {
630        return Err(VideoCompositorError::InvalidFrameRate(options.frame_rate));
631    }
632    Ok(())
633}
634
635/// Thin adapters over the shared, backend-agnostic logic in
636/// [`super::video_layer`] — translate its [`VideoLayerError`] into this
637/// backend's own [`VideoCompositorError`] variants so every existing call
638/// site/error consumer here keeps seeing the same error shape it always
639/// has.
640fn map_layer_error(error: VideoLayerError) -> VideoCompositorError {
641    match error {
642        VideoLayerError::InvalidDimensions { width, height } => {
643            VideoCompositorError::InvalidLayerDimensions { width, height }
644        }
645        VideoLayerError::InvalidOpacity(opacity) => VideoCompositorError::InvalidOpacity(opacity),
646        VideoLayerError::InvalidInputDimensions { width, height } => {
647            VideoCompositorError::InvalidInputDimensions { width, height }
648        }
649        VideoLayerError::ScaledLayerTooLarge { width, height } => {
650            VideoCompositorError::ScaledLayerTooLarge { width, height }
651        }
652    }
653}
654
655fn validate_layer(layer: VideoLayer) -> std::result::Result<(), VideoCompositorError> {
656    video_layer::validate_layer(layer).map_err(map_layer_error)
657}
658
659fn validate_rect(rect: VideoRect) -> std::result::Result<(), VideoCompositorError> {
660    video_layer::validate_rect(rect).map_err(map_layer_error)
661}
662
663fn validate_opacity(opacity: f32) -> std::result::Result<(), VideoCompositorError> {
664    video_layer::validate_opacity(opacity).map_err(map_layer_error)
665}
666
667fn layer_geometry(
668    source_width: u32,
669    source_height: u32,
670    rect: VideoRect,
671    fit: VideoFit,
672) -> std::result::Result<LayerGeometry, VideoCompositorError> {
673    video_layer::layer_geometry(source_width, source_height, rect, fit).map_err(map_layer_error)
674}
675
676fn fill_background(frame: &mut ffmpeg::frame::Video, color: Color) {
677    let width = frame.width() as usize;
678    let height = frame.height() as usize;
679    let stride = frame.stride(0);
680    let data = frame.data_mut(0);
681    for row in 0..height {
682        for pixel in data[row * stride..row * stride + width * 4].chunks_exact_mut(4) {
683            pixel.copy_from_slice(&[color.blue, color.green, color.red, 255]);
684        }
685    }
686}
687
688fn blend_bgra(
689    destination: &mut ffmpeg::frame::Video,
690    source: &ffmpeg::frame::Video,
691    geometry: LayerGeometry,
692    opacity: f32,
693) {
694    let output_width = i64::from(destination.width());
695    let output_height = i64::from(destination.height());
696    let clip_left = i64::from(geometry.clip.x).max(0);
697    let clip_top = i64::from(geometry.clip.y).max(0);
698    let clip_right =
699        (i64::from(geometry.clip.x) + i64::from(geometry.clip.width)).min(output_width);
700    let clip_bottom =
701        (i64::from(geometry.clip.y) + i64::from(geometry.clip.height)).min(output_height);
702    let left = geometry.image_x.max(clip_left);
703    let top = geometry.image_y.max(clip_top);
704    let right = (geometry.image_x + i64::from(geometry.image_width)).min(clip_right);
705    let bottom = (geometry.image_y + i64::from(geometry.image_height)).min(clip_bottom);
706    if left >= right || top >= bottom {
707        return;
708    }
709
710    let source_stride = source.stride(0);
711    let destination_stride = destination.stride(0);
712    let source_data = source.data(0);
713    let destination_data = destination.data_mut(0);
714    for output_y in top..bottom {
715        let source_y = (output_y - geometry.image_y) as usize;
716        let destination_y = output_y as usize;
717        for output_x in left..right {
718            let source_x = (output_x - geometry.image_x) as usize;
719            let destination_x = output_x as usize;
720            let source_offset = source_y * source_stride + source_x * 4;
721            let destination_offset = destination_y * destination_stride + destination_x * 4;
722            let source_pixel = &source_data[source_offset..source_offset + 4];
723            let destination_pixel =
724                &mut destination_data[destination_offset..destination_offset + 4];
725            let alpha = (f32::from(source_pixel[3]) / 255.0) * opacity;
726            let inverse = 1.0 - alpha;
727            for channel in 0..3 {
728                destination_pixel[channel] = (f32::from(source_pixel[channel]) * alpha
729                    + f32::from(destination_pixel[channel]) * inverse)
730                    .round()
731                    .clamp(0.0, 255.0) as u8;
732            }
733            destination_pixel[3] = 255;
734        }
735    }
736}
737
738#[cfg(test)]
739mod tests {
740    use std::{
741        sync::{
742            Mutex as StdMutex,
743            atomic::{AtomicBool, Ordering as AtomicOrdering},
744        },
745        thread,
746    };
747
748    use super::*;
749
750    struct CapturingSink {
751        pp_log: PpLog,
752        received: Arc<StdMutex<Vec<MediaBuffer>>>,
753    }
754
755    impl Element for CapturingSink {
756        fn name(&self) -> Arc<str> {
757            "capture".into()
758        }
759
760        fn element_type(&self) -> ElementType {
761            ElementType::Other
762        }
763
764        fn pp_log(&self) -> &PpLog {
765            &self.pp_log
766        }
767
768        fn pp_log_mut(&mut self) -> &mut PpLog {
769            &mut self.pp_log
770        }
771    }
772
773    impl Sink for CapturingSink {
774        fn consume(&mut self, buf: MediaBuffer) -> Result<()> {
775            self.received.lock().unwrap().push(buf);
776            Ok(())
777        }
778
779        fn control(&mut self, _msg: ControlMsg) -> Result<()> {
780            Ok(())
781        }
782    }
783
784    fn options(width: u32, height: u32) -> VideoCompositorOptions {
785        VideoCompositorOptions {
786            width,
787            height,
788            frame_rate: ffmpeg::Rational::new(30, 1),
789            background: Color::BLACK,
790        }
791    }
792
793    fn solid_frame(
794        width: u32,
795        height: u32,
796        color: Color,
797    ) -> Arc<UnboundObjectPoolRef<ffmpeg::frame::Video>> {
798        let pool = UnboundObjectPool::new(
799            0,
800            move || ffmpeg::frame::Video::new(ffmpeg::format::Pixel::BGRA, width, height),
801            |_| {},
802        );
803        let mut frame = pool.get();
804        fill_background(&mut frame, color);
805        Arc::new(frame)
806    }
807
808    fn pixel(frame: &ffmpeg::frame::Video, x: usize, y: usize) -> [u8; 4] {
809        let offset = y * frame.stride(0) + x * 4;
810        frame.data(0)[offset..offset + 4].try_into().unwrap()
811    }
812
813    fn input(
814        handle: &VideoCompositorHandle,
815        name: &str,
816        layer: VideoLayer,
817    ) -> (Box<dyn Sink>, VideoLayerHandle) {
818        let input = handle.add_source(name, layer).unwrap().unwrap();
819        (input.sink, input.layer)
820    }
821
822    #[test]
823    fn composes_inputs_in_z_order_and_preserves_output_contract() {
824        let (mut compositor, handle) = VideoCompositor::new("compositor", options(4, 4)).unwrap();
825        let mut background = VideoLayer::new(VideoRect::new(0, 0, 4, 4));
826        background.fit = VideoFit::Stretch;
827        let (mut red_sink, _) = input(&handle, "red", background);
828        let mut overlay = VideoLayer::new(VideoRect::new(1, 1, 2, 2));
829        overlay.z_index = 1;
830        overlay.fit = VideoFit::Stretch;
831        let (mut blue_sink, _) = input(&handle, "blue", overlay);
832        red_sink
833            .consume(MediaBuffer::Video(solid_frame(4, 4, Color::new(255, 0, 0))))
834            .unwrap();
835        blue_sink
836            .consume(MediaBuffer::Video(solid_frame(2, 2, Color::new(0, 0, 255))))
837            .unwrap();
838
839        let frame = compositor.compose_frame().unwrap();
840        assert_eq!(frame.format(), ffmpeg::format::Pixel::BGRA);
841        assert_eq!((frame.width(), frame.height()), (4, 4));
842        assert_eq!(frame.pts(), Some(0));
843        assert_eq!(pixel(&frame, 0, 0), [0, 0, 255, 255]);
844        assert_eq!(pixel(&frame, 1, 1), [255, 0, 0, 255]);
845    }
846
847    #[test]
848    fn layer_handle_moves_blends_and_hides_a_live_source() {
849        let (mut compositor, handle) = VideoCompositor::new("compositor", options(3, 1)).unwrap();
850        let layer = VideoLayer::new(VideoRect::new(0, 0, 1, 1));
851        let (mut sink, layer_handle) = input(&handle, "white", layer);
852        sink.consume(MediaBuffer::Video(solid_frame(1, 1, Color::WHITE)))
853            .unwrap();
854
855        layer_handle.set_rect(VideoRect::new(1, 0, 1, 1)).unwrap();
856        layer_handle.set_opacity(0.5).unwrap();
857        let blended = compositor.compose_frame().unwrap();
858        assert_eq!(pixel(&blended, 0, 0), [0, 0, 0, 255]);
859        assert_eq!(pixel(&blended, 1, 0), [128, 128, 128, 255]);
860
861        layer_handle.set_visible(false).unwrap();
862        let hidden = compositor.compose_frame().unwrap();
863        assert_eq!(pixel(&hidden, 1, 0), [0, 0, 0, 255]);
864        assert_eq!(hidden.pts(), Some(1));
865    }
866
867    #[test]
868    fn input_keeps_only_the_latest_frame() {
869        let (mut compositor, handle) = VideoCompositor::new("compositor", options(1, 1)).unwrap();
870        let (mut sink, _) = input(
871            &handle,
872            "latest",
873            VideoLayer::new(VideoRect::new(0, 0, 1, 1)),
874        );
875        sink.consume(MediaBuffer::Video(solid_frame(1, 1, Color::new(255, 0, 0))))
876            .unwrap();
877        sink.consume(MediaBuffer::Video(solid_frame(1, 1, Color::new(0, 255, 0))))
878            .unwrap();
879
880        let frame = compositor.compose_frame().unwrap();
881        assert_eq!(pixel(&frame, 0, 0), [0, 255, 0, 255]);
882    }
883
884    #[test]
885    fn frame_replacement_and_composition_run_concurrently() {
886        let (mut compositor, handle) = VideoCompositor::new("compositor", options(1, 1)).unwrap();
887        let (mut sink, _) = input(&handle, "live", VideoLayer::new(VideoRect::new(0, 0, 1, 1)));
888        let red = solid_frame(1, 1, Color::new(255, 0, 0));
889        let green = solid_frame(1, 1, Color::new(0, 255, 0));
890        let done = Arc::new(AtomicBool::new(false));
891        let producer_done = done.clone();
892        let producer = thread::spawn(move || {
893            for index in 0..2_000 {
894                let frame = if index % 2 == 0 {
895                    red.clone()
896                } else {
897                    green.clone()
898                };
899                sink.consume(MediaBuffer::Video(frame)).unwrap();
900            }
901            // Make the final observable value deterministic after the
902            // concurrent replacement phase ends.
903            sink.consume(MediaBuffer::Video(green)).unwrap();
904            producer_done.store(true, AtomicOrdering::Release);
905        });
906
907        while !done.load(AtomicOrdering::Acquire) {
908            let frame = compositor.compose_frame().unwrap();
909            assert!(matches!(
910                pixel(&frame, 0, 0),
911                [0, 0, 0, 255] | [0, 0, 255, 255] | [0, 255, 0, 255]
912            ));
913        }
914        producer.join().unwrap();
915        let final_frame = compositor.compose_frame().unwrap();
916        assert_eq!(pixel(&final_frame, 0, 0), [0, 255, 0, 255]);
917    }
918
919    #[test]
920    fn replacing_a_name_invalidates_old_sink_and_layer_handle() {
921        let (mut compositor, handle) = VideoCompositor::new("compositor", options(1, 1)).unwrap();
922        let layer = VideoLayer::new(VideoRect::new(0, 0, 1, 1));
923        let (mut old_sink, old_layer) = input(&handle, "camera", layer);
924        let (mut new_sink, _) = input(&handle, "camera", layer);
925        assert!(matches!(
926            old_layer.set_visible(false),
927            Err(VideoCompositorError::SourceRemoved)
928        ));
929        old_sink
930            .consume(MediaBuffer::Video(solid_frame(1, 1, Color::new(255, 0, 0))))
931            .unwrap();
932        new_sink
933            .consume(MediaBuffer::Video(solid_frame(1, 1, Color::new(0, 0, 255))))
934            .unwrap();
935
936        let frame = compositor.compose_frame().unwrap();
937        assert_eq!(pixel(&frame, 0, 0), [255, 0, 0, 255]);
938        assert_eq!(handle.source_count(), 1);
939    }
940
941    #[test]
942    fn stop_removes_only_the_current_registration() {
943        let (_compositor, handle) = VideoCompositor::new("compositor", options(1, 1)).unwrap();
944        let layer = VideoLayer::new(VideoRect::new(0, 0, 1, 1));
945        let (mut old_sink, _) = input(&handle, "camera", layer);
946        let (_new_sink, _) = input(&handle, "camera", layer);
947        old_sink.control(ControlMsg::Stop).unwrap();
948        assert_eq!(handle.source_count(), 1);
949
950        handle.remove_source("camera");
951        assert_eq!(handle.source_count(), 0);
952    }
953
954    #[test]
955    fn contain_and_cover_preserve_aspect_ratio() {
956        let rect = VideoRect::new(10, 20, 100, 100);
957        let contain = layer_geometry(160, 90, rect, VideoFit::Contain).unwrap();
958        assert_eq!((contain.image_width, contain.image_height), (100, 56));
959        assert_eq!((contain.image_x, contain.image_y), (10, 42));
960
961        let cover = layer_geometry(160, 90, rect, VideoFit::Cover).unwrap();
962        assert_eq!((cover.image_width, cover.image_height), (178, 100));
963        assert_eq!((cover.image_x, cover.image_y), (-29, 20));
964    }
965
966    #[test]
967    fn rejects_invalid_layers_and_non_video_buffers() {
968        let (_compositor, handle) = VideoCompositor::new("compositor", options(1, 1)).unwrap();
969        let invalid = VideoLayer {
970            opacity: 1.5,
971            ..VideoLayer::new(VideoRect::new(0, 0, 1, 1))
972        };
973        assert!(matches!(
974            handle.add_source("invalid", invalid),
975            Err(VideoCompositorError::InvalidOpacity(1.5))
976        ));
977
978        let (mut sink, _) = input(
979            &handle,
980            "valid",
981            VideoLayer::new(VideoRect::new(0, 0, 1, 1)),
982        );
983        let error = sink
984            .consume(MediaBuffer::Packet(Arc::new(ffmpeg::Packet::empty())))
985            .unwrap_err();
986        assert!(matches!(
987            error,
988            crate::Error::VideoCompositorError(VideoCompositorError::UnsupportedBuffer("Packet"))
989        ));
990    }
991
992    #[test]
993    fn pushes_fixed_format_frames_with_contiguous_pts() {
994        let (mut compositor, _) = VideoCompositor::new("compositor", options(2, 2)).unwrap();
995        let received = Arc::new(StdMutex::new(Vec::new()));
996        compositor.src_pads()[0].link(Box::new(CapturingSink {
997            received: received.clone(),
998            pp_log: element_pp_log(ElementType::Other, "capture", None),
999        }));
1000        let (bus, _) = Bus::new();
1001        compositor.push_frame(&bus).unwrap();
1002        compositor.push_frame(&bus).unwrap();
1003
1004        let received = received.lock().unwrap();
1005        let pts: Vec<_> = received
1006            .iter()
1007            .filter_map(|buffer| match buffer {
1008                MediaBuffer::Video(frame) => Some(frame.pts()),
1009                _ => None,
1010            })
1011            .collect();
1012        assert_eq!(pts, vec![Some(0), Some(1)]);
1013    }
1014
1015    struct TimestampSink {
1016        pp_log: PpLog,
1017        tx: crossbeam_channel::Sender<Instant>,
1018    }
1019
1020    impl Element for TimestampSink {
1021        fn name(&self) -> Arc<str> {
1022            "timestamp-recorder".into()
1023        }
1024        fn element_type(&self) -> ElementType {
1025            ElementType::Other
1026        }
1027        fn pp_log(&self) -> &PpLog {
1028            &self.pp_log
1029        }
1030        fn pp_log_mut(&mut self) -> &mut PpLog {
1031            &mut self.pp_log
1032        }
1033    }
1034
1035    impl Sink for TimestampSink {
1036        fn consume(&mut self, buf: MediaBuffer) -> Result<()> {
1037            if matches!(buf, MediaBuffer::Video(_)) {
1038                let _ = self.tx.send(Instant::now());
1039            }
1040            Ok(())
1041        }
1042        fn control(&mut self, _msg: ControlMsg) -> Result<()> {
1043            Ok(())
1044        }
1045    }
1046
1047    /// Regression test: `VideoCompositor::run` never folded
1048    /// `ControlOutcome::paused_for` back into `next_due` — a `Pause` let
1049    /// real time blow straight past the stale deadline, so the loop
1050    /// iteration right after `Resume` always found `next_due` already in
1051    /// the past and pushed immediately, resetting the output cadence's
1052    /// phase to the resume instant instead of preserving wherever it was
1053    /// before the freeze. A slow 10fps (100ms/frame) rate keeps the
1054    /// expected gap (near-zero vs. near-one-interval) well clear of
1055    /// scheduling jitter. Pausing is triggered synchronously right after a
1056    /// frame is observed, so `next_due` is a known ~100ms away the instant
1057    /// `Pause` is drained (`run`'s own control check only happens at the
1058    /// top of the loop, after that deadline has already been advanced).
1059    #[test]
1060    fn resuming_after_a_pause_preserves_output_phase() {
1061        use crate::pipeline::Pipeline;
1062
1063        let (tx, rx) = crossbeam_channel::unbounded();
1064        let sink = TimestampSink {
1065            tx,
1066            pp_log: element_pp_log(ElementType::Other, "timestamp-recorder", None),
1067        };
1068        let (compositor, _handle) = VideoCompositor::new(
1069            "compositor",
1070            VideoCompositorOptions {
1071                frame_rate: ffmpeg::Rational::new(10, 1),
1072                ..options(2, 2)
1073            },
1074        )
1075        .unwrap();
1076
1077        let pipeline = Pipeline::new("phase-test", compositor, |source, ctx| {
1078            let branch = ctx.branch().to(Box::new(sink))?;
1079            ctx.attach(source, 0, branch)?;
1080            Ok(())
1081        })
1082        .expect("test pipeline wiring must succeed");
1083
1084        pipeline.run();
1085        // Warm up, then pause the instant a frame is observed — `next_due`
1086        // is then a known one interval away.
1087        for _ in 0..2 {
1088            rx.recv_timeout(Duration::from_millis(500))
1089                .expect("expected steady frames before pausing");
1090        }
1091        pipeline.pause();
1092        thread::sleep(Duration::from_millis(500));
1093
1094        let resumed_at = Instant::now();
1095        pipeline.resume();
1096        let first_after_resume = rx
1097            .recv_timeout(Duration::from_millis(500))
1098            .expect("expected a frame after resume");
1099        pipeline.stop();
1100        pipeline.bus().log_events();
1101
1102        let gap = first_after_resume.saturating_duration_since(resumed_at);
1103        assert!(
1104            gap >= Duration::from_millis(50),
1105            "expected the post-pause frame to land close to a full 100ms \
1106             interval after resume (phase preserved from before the \
1107             pause), not almost immediately (phase reset to the resume \
1108             instant): got {gap:?}"
1109        );
1110    }
1111}
media_pp\elements\source\compositor/video_compositor.rs

media_pp\elements\source\compositor/
video_compositor.rs
